package kafka.examples.my;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 *
 */
public class MyProducer {

    private static final String TOPIC = "topic2";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 64*1024*1024); //producer缓冲区大小
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024); // 16k
        props.put(ProducerConfig.LINGER_MS_CONFIG, 100); //send()一条消息, 100ms内还没凑成一个batch, 立刻发出去
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10*1024*1024); //每次发给broker的请求数据包最大大小, 10MB
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.RETRIES_CONFIG, 3); //失败重试次数, 一般3-5次
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 500); //每次重试间隔ms
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");  //消息压缩算法

        KafkaProducer<Integer, String> producer = new KafkaProducer<>(props);

        int count = 2;
        for (int i = 0; i < count; i++) {
            long startTime = System.currentTimeMillis();
            String message = "Message_" + startTime;
            int key = i;
            producer.send(new ProducerRecord<>(TOPIC, key, message), new DemoCallBack(startTime, key, message));
        }

        Sleep.sleep(TimeUnit.SECONDS, 61);
    }

}

class DemoCallBack implements Callback {

    private final long startTime;
    private final int key;
    private final String message;

    public DemoCallBack(long startTime, int key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }

    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            System.out.println(
                    "message(" + key + ", " + message + ") sent to " +
                        "partition(" + metadata.partition() + "), " +
                        "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
        } else {
            exception.printStackTrace();
        }
    }
}